# 输出文件解析
# sales_top_products.csv 文件
# title：商品名称（来自dwd_products.csv的title字段）
# deal_num：总销量（对同一商品deal_num字段的求和）
# price：该商品的平均售价（原始price字段的平均值）
# 业务意义：展示当前最畅销的10个商品及其销售表现，用于识别爆款商品。例如：
# 商品A以299元均价卖出1500件 → 主力创收商品
# 商品B以199元卖出1200件 → 高销量平价商品

# product_association_rules.csv 文件
# antecedents：前置商品（触发购买的商品）
# consequents：后置商品（被连带购买的商品）
# support：组合出现频率（0.15表示15%的订单包含该组合）
# confidence：关联强度（0.78表示买手机的客户有78%会买耳机）
# lift：相关性系数（>1表示正相关）


import pandas as pd
import numpy as np
from mlxtend.frequent_patterns import apriori, association_rules
import os
import time


class TaobaoAnalyzer:
    def __init__(self):
        # 配置基础路径（根据图片调整）
        self.base_path = "D:/DataWarehouse"
        self.output_path = os.path.join(self.base_path, "ana_out")
        os.makedirs(self.output_path, exist_ok=True)

        # 数据文件路径（根据图片中的实际文件名调整）
        self.data_paths = {
            'products': os.path.join(self.base_path, "outputs/dwd/dwd_products.csv"),
            'province_sales': os.path.join(self.base_path, "outputs/dws/dws_province_sales.csv"),
            'shop_stats': os.path.join(self.base_path, "outputs/dws/dws_shop_stats.csv"),
            'raw_data': os.path.join(self.base_path, "outputs/ods/odsRaw_data.csv")
        }

        # 加载数据
        self.load_data()

    def load_data(self):
        """加载所有数据文件（移除parse_dates参数）"""
        try:
            # 读取产品数据（不再假设有load_time列）
            self.products = pd.read_csv(self.data_paths['products'])

            # 读取其他数据
            self.province_sales = pd.read_csv(self.data_paths['province_sales'])
            self.shop_stats = pd.read_csv(self.data_paths['shop_stats'])

            # 原始数据可选加载
            if os.path.exists(self.data_paths['raw_data']):
                # 尝试两种格式
                try:
                    self.raw_data = pd.read_csv(self.data_paths['raw_data'])
                except:
                    self.raw_data = pd.read_parquet(self.data_paths['raw_data'].replace('.csv', '.parquet'))
            else:
                self.raw_data = None

        except Exception as e:
            print(f"数据加载失败: {str(e)}")
            print("请检查以下文件是否存在:")
            for name, path in self.data_paths.items():
                print(f"{name}: {path}")
            raise

    def sales_analysis(self):
        """销售分析模块（不再依赖时间字段）"""
        results = {}

        try:
            # 1. Top10畅销商品（使用图片中显示的deal_num字段）
            if 'deal_num' in self.products.columns:
                top_products = self.products.groupby('title').agg({
                    'deal_num': 'sum',
                    'price': 'mean'
                }).sort_values('deal_num', ascending=False).head(10)
                results['top_products'] = top_products

            # 2. 地区销售分布
            if 'total_sales' in self.province_sales.columns:
                region_dist = self.province_sales.sort_values('total_sales', ascending=False)
                results['region_dist'] = region_dist

            # 保存结果
            for name, df in results.items():
                df.to_csv(os.path.join(self.output_path, f"sales_{name}.csv"), encoding='utf-8-sig')

            return results

        except Exception as e:
            print(f"销售分析失败: {str(e)}")
            return {}

    def user_behavior_analysis(self):
        """用户行为分析模块"""
        results = {}

        try:
            # 1. 用户复购率计算（检查必要字段）
            if all(col in self.products.columns for col in ['user_id', 'order_id']):
                user_orders = self.products.groupby('user_id')['order_id'].nunique()
                repurchase_rate = (user_orders[user_orders > 1].count() / user_orders.count()) * 100
                results['repurchase_rate'] = pd.DataFrame({
                    'metric': ['repurchase_rate'],
                    'value': [repurchase_rate]
                })

            # 2. 购物车放弃率（需要原始数据支持）
            if self.raw_data is not None and all(col in self.raw_data.columns for col in ['event_type', 'purchased']):
                cart_events = self.raw_data[self.raw_data['event_type'] == 'cart']
                abandoned = cart_events[cart_events['purchased'] == False].shape[0]
                total = cart_events.shape[0]
                abandon_rate = (abandoned / total) * 100 if total > 0 else 0
                results['cart_abandon_rate'] = pd.DataFrame({
                    'metric': ['cart_abandon_rate'],
                    'value': [abandon_rate]
                })

            # 保存结果
            for name, df in results.items():
                df.to_csv(os.path.join(self.output_path, f"user_{name}.csv"), encoding='utf-8-sig')

            return results

        except Exception as e:
            print(f"用户行为分析失败: {str(e)}")
            return {}

    def product_association(self, min_support=0.05, min_confidence=0.3):
        """商品关联分析（Apriori算法）"""
        try:
            # 检查必要字段
            if not all(col in self.products.columns for col in ['order_id', 'product_id']):
                print("缺少必要的order_id或product_id字段")
                return pd.DataFrame()

            # 准备交易数据
            transaction_data = self.products.groupby('order_id')['product_id'].apply(list)

            # 转换为one-hot编码格式
            transactions = transaction_data.explode().to_frame()
            transactions['value'] = 1
            transactions = transactions.pivot_table(
                index='order_id',
                columns='product_id',
                values='value',
                fill_value=0
            ).astype(bool)

            # Apriori算法
            frequent_itemsets = apriori(
                transactions,
                min_support=min_support,
                use_colnames=True
            )

            # 关联规则
            rules = association_rules(
                frequent_itemsets,
                metric="confidence",
                min_threshold=min_confidence
            ).sort_values('confidence', ascending=False)

            # 保存结果
            rules.to_csv(
                os.path.join(self.output_path, "product_association_rules.csv"),
                index=False,
                encoding='utf-8-sig'
            )
            return rules

        except Exception as e:
            print(f"关联分析失败: {str(e)}")
            return pd.DataFrame()

    def run_all_analysis(self):
        """执行全部分析"""
        print(">>> 开始执行分析任务")
        start_time = time.time()

        results = {
            'sales': self.sales_analysis(),
            'user_behavior': self.user_behavior_analysis(),
            'association': self.product_association()
        }

        print(f">>> 分析完成，耗时: {time.time() - start_time:.2f}秒")
        print(f"结果已保存到: {self.output_path}")
        return results


if __name__ == "__main__":
    analyzer = TaobaoAnalyzer()
    analyzer.run_all_analysis()